package com.oct.ywsc.job;

import com.oct.ywsc.ITaskScheduler;
import com.oct.ywsc.core.OptTypeEnum;
import com.oct.ywsc.core.bean.MsgBean;
import com.oct.ywsc.core.bean.RuleConfig;
import com.oct.ywsc.core.config.KafkaConsumerConfig;
import com.oct.ywsc.core.config.ThreadConfig;
import com.oct.ywsc.core.entity.GzConfigEntity;
import com.oct.ywsc.core.service.GzConfigService;
import com.oct.ywsc.job.kafka.IClient;
import com.oct.ywsc.job.kafka.MQClient;
import com.oct.ywsc.job.schedule.TransJobExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;


/**
 * 消息转换任务调度器
 */
@Slf4j
@Component
public class TransTaskScheduler implements ITaskScheduler {

    /**
     * 是否可以接受消息
     */
    public static boolean canAcceptable = true;

    public static boolean reflushRuleManageing = false;  // 记录是否正在刷新规则管理器


    public static boolean stoping = false;  // 停止服务标记


    @Autowired
    private GzConfigService gzConfigService;

    @Resource
    @Qualifier("customerProps")
    private Map<String, Object> customerProps;

    @Autowired
    private KafkaConsumerConfig kafkaConsumerConfig;

    @Autowired
    private KafkaTemplate kafkaTemplate;

    private CustomerKafkaProducer customerKafkaProducer ;

    @Autowired
    private ThreadConfig threadConfig;

//    @Autowired
//    private JdbcTemplate jdbcTemplate;

    @Autowired
    private DataSource dataSource;

    RuleManage ruleManage ;


    // 消息处理调度
    private TransJobExecutor transJobExecutor;


    KafkaConsumer<String, String> consumer;    // kafka消费端


    private IClient client ;

    /**
     * 系统启动后初始化运行环境
     * 自动加载规则配置类
     */
    @PostConstruct
//    @EventListener(ApplicationReadyEvent.class)
    public void onApplicationReady(){


        // 初始化kafka消息生产器
        this.customerKafkaProducer = new CustomerKafkaProducer(kafkaTemplate);

        // 创建消息转换调度器
        transJobExecutor = new TransJobExecutor(threadConfig);

        // 刷新规则管理器
        this.reflushRuleManage();

//        //  创建 Kafka 消费者
        client = new MQClient(customerProps,kafkaConsumerConfig.getTopicNames());
        client.init();

    }


    @PreDestroy
    public void onShutdown(){

        // 设置停止服务标记
        TransTaskScheduler.stoping = true;  // 标记停止服务
//        do{
//            log.info(String.format("待处理任务数[%s]，请等待处理完成。。", transJobExecutor.jobSize()));
//        }while(this.transJobExecutor.jobSize()>0);

        log.info("kafka监听程序关闭");
        client.close();
    }


    public long jobSize(){
        return transJobExecutor.jobSize();
    }


    /**
     * 开启监听kafka 消息
     * 定时任务 每1分钟钟执行一次
     */
    @Scheduled(fixedDelay = 60000,initialDelay = 0)
    public void processing(){


        if (TransTaskScheduler.stoping){
            log.info("正在停止服务，暂停接收kafka消息，待处理任务数量："+transJobExecutor.jobSize());
            return ;
        }

        log.info("kafka监听程序开始");
        if (reflushRuleManageing){
            log.info("规则管理器正在刷新，暂停接收kafka消息");
            return;
        }



        // 判断当前消息队列是否可以接受消息
        if (!this.transJobExecutor.canAcceptable(this.kafkaConsumerConfig.getMaxPollRecords())
        && !TransTaskScheduler.canAcceptable){
            log.warn(String.format("当前待处理对象数量%s，已达最大容量，需稍等", transJobExecutor.jobSize()));
            // 判断是否可以接受消息，不可接收则跳过,则等待5秒后继续监听消息
//            this.transJobExecutor.canAcceptable(this.kafkaConsumerConfig.getMaxPollRecords());
            return;
        }

        // 4. 持续监听消息
        List<MsgBean> records = client.poll(Duration.ofMillis(kafkaConsumerConfig.getMaxPollIntervalMs()));
        if (records != null){
            for ( MsgBean record : records) {

                JdbcTemplate jt = new JdbcTemplate(dataSource);

                MsgTransTask job = new MsgTransTask(record, ruleManage,
                        jt, customerKafkaProducer);
                transJobExecutor.add(job);
            }
            // 提交消费进度
            client.commitSync();
        }

    }

    @Override
    public void reflushRuleManage() {


        // 判断是否有任务待执行，没有任务则从数据库中加载规则配置
        if (this.transJobExecutor.jobSize() <=0) {

            List<GzConfigEntity> list = gzConfigService.list();
            // 构建 RouteManage
            ruleManage = new RuleManage();
            for (GzConfigEntity item : list) {

                // 解析数据库中配置规则，根据opttype 拆分成 RouleConfig
                List<TransRule> transRuleList = this.buildTransRuleList(item);
                ruleManage.addAll(transRuleList);
            }

            log.info("规则配置刷新完成");
            // 刷新完成标记刷新状态结束
            reflushRuleManageing = false;
        }else{
            log.info(String.format("转换任务剩余%s个，执行完成后开始刷新规则配置。。", transJobExecutor.jobSize()));
        }
    }


    /**
     * 根据操作类型拆分多个转换规则
     * @param entity
     * @return
     */
    private List<TransRule> buildTransRuleList(GzConfigEntity entity){

        List<TransRule> transRuleList = new ArrayList<>();
        // 判断 optype 是否包含逗号，包含逗号则需要拆分
        if (entity.getOpType().contains(",")){
            // 根据操作类型进行拆分多个操作配置对象
            String[] ot = entity.getOpType().split(",");
            for (String o : ot){
                RuleConfig ruleConfig =  new RuleConfig();
                BeanUtils.copyProperties(entity,ruleConfig);
                OptTypeEnum e = OptTypeEnum.findOptTypeEnum(o);
                ruleConfig.setOptTypeEnum(e);
                transRuleList.add(new TransRule(ruleConfig));
            }

        }else{
            RuleConfig ruleConfig =  new RuleConfig();
            BeanUtils.copyProperties(entity,ruleConfig);
            OptTypeEnum e = OptTypeEnum.findOptTypeEnum(entity.getOpType());
            ruleConfig.setOptTypeEnum(e);
            transRuleList.add(new TransRule(ruleConfig));
        }

        return transRuleList;
    }

}
